package com.hllxd.hism.star.basic.service.collect;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.TypeReference;
import com.hllxd.hism.star.basic.service.dao.redis.YarnRedisMapper;
import com.hllxd.hism.star.basic.service.exception.BusinessException;
import com.hllxd.hism.star.basic.service.model.rock.YarnApplicationEntity;
import com.hllxd.hism.star.basic.service.model.rock.YarnQuotaEntity;
import com.hllxd.hism.star.basic.service.util.TimeUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @author xiyong.lxy
 * @date 2021-8-17 10:10
 * 需要将服务放到其他模块
 */
@Component
public class CollectYarn {

    @Autowired
    private YarnRedisMapper redisMapper;

    private static final Logger logger = LoggerFactory.getLogger(CollectYarn.class);

    public static final String YARN_URL = "http://wq-yarn1-rm1.58dns.org:9088";

    static String[] QUEUES = (
            "root.offline.hdp_lbg_supin,root.offline.hdp_lbg_zhaopin"
    ).split(",");

    public static YarnApplicationEntity parseApplication(Map<String, Object> appMap) {

        YarnApplicationEntity app = new YarnApplicationEntity();
        try {
            // 任务核数
            Long allocatedVCores = Double.valueOf(appMap.getOrDefault("allocatedVCores", 100).toString()).longValue();
            // 任务内存
            Long allocatedMB = Double.valueOf(appMap.getOrDefault("allocatedMB", 10240).toString()).longValue();
            // 持续时间
            Long elapsedTime = Double.valueOf(appMap.get("elapsedTime").toString()).longValue();
            // 开始时间
            Long startedTime = Double.valueOf(appMap.get("startedTime").toString()).longValue();

            String appName = appMap.get("name").toString();
//            app.accessID = appMap.get("user").toString();
//            app.appID = appMap.get("id").toString();
//            app.jobName = appName.replaceAll("'", "").replaceAll("\"", "");
            app.applicationType = appMap.get("applicationType").toString();
            app.queue = appMap.get("queue").toString();
//            app.allocatedVCores = String.valueOf(allocatedVCores);
//            app.allocatedMB = String.valueOf(allocatedMB);
//            app.state = appMap.get("state").toString();
//            app.progress = appMap.get("progress").toString();
//            app.finalStatus = appMap.get("finalStatus").toString();
//            app.finishedTime = appMap.get("finishedTime").toString();
//            app.elapsedTime = DateUtils.formatMillisecond(elapsedTime);
//            app.startedTime = DateUtils.formatDateTime(startedTime);
//            app.createTime = DateUtils.formatDateTime(new Date());
        } catch (Exception e) {
            logger.warn("Exception: {}.", e, e);
        }
        return app;
    }

    /**
     * 获取yarn任务列表
     * https://blog.csdn.net/lumingkui1990/article/details/52175263
     * http://wq-yarn1-rm1.58dns.org:9088/ws/v1/cluster/apps?queue=root.offline.default
     *
     * @throws IOException
     */
    public static List<YarnApplicationEntity> getAppList(Map<String, String> params) throws IOException {
        // 当前时间
//        Date nowD = new Date();
//        TreeMap<String,String> paramUrl = new TreeMap<>();
//        paramUrl.put("state","running");
//        paramUrl.put("FinishedTimeBegin", (nowD.getTime() - 60000L) + "");
//        paramUrl.put("FinishedTimeEnd", nowD.getTime() + "");
        String uri = "/ws/v1/cluster/apps";
        List<YarnApplicationEntity> results = new ArrayList<>();
        String queueParam = params.get("queue");
        if (queueParam == null || queueParam.equals("")) {
            for (String queue : QUEUES) {
                Map<String, String> reqParam = new HashMap<>();
                reqParam.put("queue", queue);
                if (params.containsKey("states")) {
                    reqParam.put("states", params.get("states"));
                }
                OkHttpUtil httpUtil = OkHttpUtil.getInstance();
                String res = httpUtil.getDataStringByParam(YARN_URL + uri, reqParam);
//                String res = HttpUtils.get(YARN_URL + uri, reqParam);
                List<YarnApplicationEntity> o = getYarnApplicationEntities(res);
                results.addAll(o);
            }
        } else {
//            String res = HttpUtils.get(YARN_URL + uri, params);
            OkHttpUtil httpUtil = OkHttpUtil.getInstance();
            String res = httpUtil.getDataStringByParam(YARN_URL + uri, params);
            List<YarnApplicationEntity> o = getYarnApplicationEntities(res);
            results.addAll(o);
        }
        return results;

    }

    private static List<YarnApplicationEntity> getYarnApplicationEntities(String res) {
        JSONObject json = JSON.parseObject(res);
        JSONObject apps = json.getJSONObject("apps");
        JSONArray array = apps.getJSONArray("app");
        List<YarnApplicationEntity> o = new ArrayList<>();
        if (array != null) {
            o = array.toJavaObject(new TypeReference<List<YarnApplicationEntity>>() {
            });
        }
        for (YarnApplicationEntity entity : o) {
            if (entity.getFinishedTime() <= 0) {
                entity.setUseTime(System.currentTimeMillis() - entity.getStartedTime());
            } else {
                entity.setUseTime(entity.getFinishedTime() - entity.getStartedTime());
            }
        }
        return o;
    }

    public static List<Map<String, Object>> getAppListByQueue(String queue) {
        String uri = "/ws/v1/cluster/apps?queue=" + queue;
        try {
            return YarnRequest.getYarn(YARN_URL, uri, null, "apps,app".split(","));
        } catch (Exception e) {
            throw new BusinessException("001", "queue: " + queue);
        }
    }

    public List<YarnApplicationEntity> listYarnApp() {
        Long currTs = TimeUtil.currentTimeHllDT();
        redisMapper.setRequestLastTs("api", currTs);

        List<YarnApplicationEntity> apps = new ArrayList<>();
        for (String queue : QUEUES) {
            String appJson = redisMapper.getQueue("app", queue);
            logger.info("get app: {}", appJson);
            if (appJson == null) {
                continue;
            }
            List<YarnApplicationEntity> appList = JSON.parseArray(appJson, YarnApplicationEntity.class);
            apps.addAll(appList);
        }
        return apps;
    }

    static String OFFLINE_QUOTA = "root.offline";

    public List<YarnQuotaEntity> listYarnQuota() {
        Long currTs = TimeUtil.currentTimeHllDT();
        redisMapper.setRequestLastTs("api", currTs);
        String json = redisMapper.getQueue("quota", OFFLINE_QUOTA);
        if (json == null) {
            logger.warn("get quota is nil.");
            return null;
        }
        logger.info("get quota: {}", json);
        return JSON.parseArray(json, YarnQuotaEntity.class);
    }


    public List<YarnApplicationEntity> listYarnAppRequest() {
        try {
            for (String queue : QUEUES) {
                List<YarnApplicationEntity> apps = new ArrayList<>();
                List<Map<String, Object>> listAppMap = getAppListByQueue(queue);
                for (Map<String, Object> appMap : listAppMap) {
                    YarnApplicationEntity app = parseApplication(appMap);
                    apps.add(app);
                }
                String appJson = JSON.toJSONString(apps);
                logger.info("set app: {}, {}.", apps.size(), appJson);
                redisMapper.setQueue("app", queue, appJson);
            }

        } catch (Exception e) {
            logger.error("Exception: {}", e, e);
        }
        return null;
    }

    /**
     * 获取资源调度分配情况
     */

    public void schedulerRequest() throws IOException {
        String uri = "/ws/v1/cluster/scheduler";
        List<YarnQuotaEntity> quotas = YarnRequest.getYarnScheduler(YARN_URL, uri);
        String quotaJson = JSON.toJSONString(quotas);
        logger.info("set quota: {}, {}.", quotas.size(), quotaJson);
        redisMapper.setQueue("quota", OFFLINE_QUOTA, quotaJson);
    }

    /**
     * 获取单个yarn任务
     *
     * @param appid 对应appid
     * @return 单个app信息
     * @throws IOException
     */
    public static List<Map<String, Object>> getApplication(String appid) throws IOException {
        String uri = "/ws/v1/cluster/apps/" + appid;
        return YarnRequest.getYarnOne(YARN_URL, uri, null, "app");
    }


    public static void main(String[] args) throws Exception {
//        List<YarnQuotaEntity> data = requestScheduler();
//        for(YarnQuotaEntity app:data) {
//            logger.info("{}", JSON.toJSONString(app));
//        }
    }
}
